package com.shujia.streeam

import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.{Duration, Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object Demo1WordCount {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local[2]").setAppName("Demo1WordCount")

    val sc = new SparkContext(conf)


    /**
      *
      * 创建 streamming上下文对象  需要指定间隔时间   延迟时间
      *
      */
    val ssc = new StreamingContext(sc, Durations.seconds(5))


    /**
      * 连接socket端口创建Dstream
      *
      * 在 linux  中执行  nc  -lk 8888   绑定一个端口提供数据服务
      * yum install nc
      */
    val ds: ReceiverInputDStream[String] = ssc.socketTextStream("node1", 8888, StorageLevel.MEMORY_AND_DISK_SER)


    //相当于action算子
    //ds.print()

    /**
      * 每隔5秒执行一次
      * 只处理当前batch 数据
      */
    ds
      .flatMap(line => line.split(","))
      .map(word => (word, 1))
      .reduceByKey((x, y) => x + y) //只统计当前batch的数据
      .print()


    /**
      *
      * 启动spark streaming程序
      *
      */

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()
  }
}
